package com.heima.homeservicemanagementsystem.module;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.DeleteResponse;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import com.heima.homeservicemanagementsystem.entity.Servicedetails;
import com.heima.homeservicemanagementsystem.listeners.AbstractCanalRabbitMqMsgListener;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RequestOptions;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Description
 * @Author jcf
 * @Date 2025/4/19
 */
@Slf4j
@Component
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<Servicedetails>  {
    @Autowired
    private ElasticsearchClient client;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "canal-mq-jzo2o-foundations"),
            exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),
            key = "canal-mq-jzo2o-foundations"),
            concurrency = "1"
    )
    public void onMessage(Message message) throws Exception {
        parseMsg(message);
    }

    @Override
    public void batchSave(List<Servicedetails> data) throws IOException {
        log.info("batchSave:{}", data.get(0));
        List<BulkOperation> bulkOperationArrayList = new ArrayList<>();

// 遍历添加到 bulk 操作中
        for (Servicedetails user : data) {
            log.info("user:{}", user);
            bulkOperationArrayList.add(BulkOperation.of(o ->
                    o.index(i -> i
                            .id(user.getId())  // 设置文档的 ID
                            .document(user)    // 设置文档内容
                    )
            ));
        }

        BulkResponse bulkResponse = client.bulk(b -> b.index("servicedetails")
                .operations(bulkOperationArrayList));
    }

    @Override
    public void batchDelete(List<Long> ids) throws IOException {
        log.info("batchDelete:{}", ids.get(0));
        for (Long id : ids) {
            DeleteResponse deleteResponse = client.delete(d -> d
                    .index("servicedetails")
                    .id(id.toString())
            );
            deleteResponse.id();
        }

    }

    @Override
    public void batchUpdate(List<Servicedetails> data) throws IOException {
        log.info("batchUpdate:{}", data.get(0));
        for (Servicedetails datum : data) {
            client.update(u -> u
                            .index("servicedetails")
                            .id(datum.getId())
                            .doc(datum)
                    , Servicedetails.class);
        }
    }
}
